package cn.smileyan.demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 *
 * @author Smileyan
 */
@Slf4j
public class TaskStateUpdater extends KeyedProcessFunction<String, String, Tuple2<String, Integer>> {
    private transient ValueState<Integer> toggleState;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        // 创建状态描述符
        ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>("toggle-state", Integer.class);

        // 获取状态
        toggleState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(String value, KeyedProcessFunction<String, String, Tuple2<String, Integer>>.Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
        Integer currentState = toggleState.value();
        if (currentState == null) {
            currentState = 0;
        }

        // 更新状态
        int newState = (currentState + 1) % 2;
        toggleState.update(newState);

        // 输出更新后的状态
        out.collect(new Tuple2<>(value, newState));
    }

}
